package org.parade.driver;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;

import org.parade.Constant;
import org.parade.distributed.Id;
import org.parade.distributed.Lock;
import org.parade.distributed.Queue;
import org.parade.distributed.Thread;


public class ThreadExecutionService extends java.lang.Thread{
	private Driver driver;
	private Queue queue;
	private boolean terminated = false;
	private Lock lock;
	private Condition condition;
	
	public ThreadExecutionService(Driver driver){
		this.driver = driver;
		queue = new Queue(driver, new Id(Constant.ID_GLOBAL_THREAD_CREATION_QUEUE));
		lock = new Lock(driver, new Id(Constant.ID_GLOBAL_THREAD_CREATION_LOCK));
		condition = lock.newCondition();
	}
	
	public void terminate(){
		terminated = true;
		add(null);
	}
	
	/**
	 * Producer
	 * 
	 * @param t
	 */
	public void add(org.parade.distributed.Thread t){
		lock.lock();
		try{
			queue.put(t);
			condition.signal();
		} catch(Exception e){
			e.printStackTrace();
		} finally {
			lock.unlock();
		}
	}
	
	/**
	 * Consumer
	 */
	@Override
	public void run(){
		while(terminated==false){
			lock.lock();
			try{
				condition.await(10000, TimeUnit.MILLISECONDS);
				if(queue.isEmpty()) continue;
				
				Thread t = (org.parade.distributed.Thread) queue.remove();
				if(t==null) continue;
				t.initialize(driver, t.getDistributedObjectId());
				Executor e = new Executor(t);
				e.start();
			} catch(Exception e){
				e.printStackTrace();
			} finally{
				lock.unlock();
			}
		}
	}
	
	
	private class Executor extends java.lang.Thread{

		private Thread thread;

		public Executor(org.parade.distributed.Thread thread){
			this.thread = thread;
		}
		
		public void run(){
			try{
				thread.onStarted();
				thread.run();
			} catch(Exception e){
			} finally{
				thread.onCompleted();
			}
		}
	}
}
